package consumer

import (
	"encoding/json"
	"errors"
	"github.com/Shopify/sarama"
	"github.com/sirupsen/logrus"
	"log"

	"kafka_go/common"
)

type EventHandler struct {
}

func (e EventHandler) Setup(session sarama.ConsumerGroupSession) error {
	return nil
}

func (e EventHandler) Cleanup(session sarama.ConsumerGroupSession) error {
	return nil
}

func (e EventHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for msg := range claim.Messages() {
		var data common.KafkaMsg
		if err := json.Unmarshal(msg.Value, &data); err != nil {
			return errors.New("failed to unmarshal message err is " + err.Error())
		}
		// 操作数据，改用打印
		log.Print("consumerClaim data is ", data)
		if err := processData(data); err != nil {
			logrus.Error("processData failed, err is ", err.Error())
			return err
		}
		// 处理消息成功后标记为处理, 然后会自动提交
		session.MarkMessage(msg, "")
	}
	return nil
}

type parseFunc func(id int, data string) error //// 2. 根据id修改结构体的reqState字段 填充对应子结构体res的结果

var parseFuncs = map[string]parseFunc{
	// 注册parse方法 key为ReqName(前端的路由名称)
	"fakeDetect": fakeDetectHandler,
}

func processData(req common.KafkaMsg) (err error) {
	// 1. 根据reqName 调用不同方法 传入request_id
	var v parseFunc
	var ok bool
	if v, ok = parseFuncs[req.ReqName]; !ok {
		return errors.New("kafka parse func key Not Found")
	}
	if err = v(req.RequestID, req.Data); err != nil {
		return errors.New("kafka parse func err")
	}
	return setReqStateFinish(req.ID)

}

// 3. 处理完毕，根据 ID 修改req_msg中 stata字段
func setReqStateFinish(id uint) (err error) {
	var req common.ReqMessage
	err = common.GetDB().First(&req, id).Error
	req.ReqState = 2
	common.GetDB().Save(&req)
	return
}
